package com.atguigu.flink.chapter11.window;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2022/11/1 09:34
 */
public class Flink03_Window_GroupSet {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<WaterSensor> stream = env.fromElements(
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s1", 1000L, 20),
            new WaterSensor("s1", 3000L, 30),
            new WaterSensor("s1", 4000L, 40)
        ).assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((ws, ts) -> ws.getTs())
        );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts"), $("vc"), $("et").rowtime());
        tEnv.createTemporaryView("sensor", table);
    
    
    
       /* tEnv.sqlQuery("select" +
                          " window_start, window_end, id,ts, " +
                          " sum(vc) vc_sum " +
                          "from table( tumble(table sensor, descriptor(et), interval '5' second ) ) " +
                          "group by window_start, window_end, grouping sets( (id, ts), (id), () )")  // window_start, window_end 务必要出现在 group by 中
            .execute()
            .print();*/
        
        // (id ts)  (id)  ()  上钻
        /*tEnv.sqlQuery("select" +
                          " window_start, window_end, id,ts, " +
                          " sum(vc) vc_sum " +
                          "from table( tumble(table sensor, descriptor(et), interval '5' second ) ) " +
                          "group by window_start, window_end, rollup(id, ts)")  // window_start, window_end 务必要出现在 group by 中
            .execute()
            .print();*/
        
        // id ts:  (id ts), (id), (ts), ()
        tEnv.sqlQuery("select" +
                          " window_start, window_end, id,ts, " +
                          " sum(vc) vc_sum " +
                          "from table( tumble(table sensor, descriptor(et), interval '5' second ) ) " +
                          "group by window_start, window_end, cube(id, ts)")  // window_start, window_end 务必要出现在 group by 中
            .execute()
            .print();
 
        
    }
}
/*

select
    a,b,c,
    sum()
from t
group by a, b, c

union

select
    a,b,""
    sum()
from t
group by a, b
union
select
    a,"",""
    sum()
from t
group by a


select
    a,b,c,
    sum()
from t
group by grouping sets ( (a,b,c), (a,b), (a))

等价于:




 */